package com.atguigu.flink.sql.join;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by Smexy on 2023/4/14
 *
 *  普通(不开窗)连接。
 *
 *  windowJoin(窗口连接)。用的少了。
 */
public class Demo1_CommonJoin
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);
        tableEnvironment.getConfig().getConfiguration().setLong("table.exec.state.ttl",10 * 1000);

        env.setParallelism(1);


        SingleOutputStreamOperator<WaterSensor> ds1 = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction());

        SingleOutputStreamOperator<WaterSensor> ds2 = env
            .socketTextStream("hadoop102", 8889)
            .map(new WaterSensorMapFunction());

        Schema schema = Schema.newBuilder()
                              .column("id", "STRING")
                              .column("ts", "BIGINT")
                              .column("vc", "INT")
                              .columnByExpression("pt", "proctime()")
                              .columnByExpression("et", "TO_TIMESTAMP_LTZ(ts,3)")
                              .build();
        //从流中获取时间属性
        tableEnvironment.createTemporaryView("t1",ds1,schema);
        tableEnvironment.createTemporaryView("t2",ds2,schema);


        /*
            常规join，双方的数据都会缓存在状态中。
                    可以设置状态的自动清理时间，将过时的状态删除，避免状态过大，拖累ck时间。

         */
        tableEnvironment
                        //.sqlQuery("select t1.id,t1.vc ,t2.id,t2.vc from t1 join t2 on t1.id = t2.id ")
                        .sqlQuery("select t1.id,t1.vc ,t2.id,t2.vc from t1 left join t2 on t1.id = t2.id ")
                        .execute()
                        .print();



        env.execute();

    }
}
